package com.dky.satell.rocketmq.consumer;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.dky.satell.mapper.MessageConfigMapper;
import com.dky.satell.rocketmq.consumer.service.RocketMQListener;
import com.dky.satell.entity.MessageConfig;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author lihaobin
 * @version 1.0
 * @date 2021/4/19 18:17
 */
@Slf4j
@Component
public class RocketMQConsumerInit {

    @Resource
    private MessageConfigMapper messageConfigMapper;

    @Getter
    private static ConcurrentHashMap<String, MessageConfig> topicMessageConfigMap = new ConcurrentHashMap<>();

    private Map<Integer, Consumer> consumerMap = new HashMap<>();

    @PostConstruct
    public void init() {
        log.info("[start]初始化启动RocketMQ消费者！");
        log.info("获取所有RocketMQ订阅信息！");
        List<MessageConfig> messageConfigList = messageConfigMapper.getAllRocketConfig();
        if(CollectionUtils.isEmpty(messageConfigList)) {
            return;
        }
        for(MessageConfig messageConfig : messageConfigList) {
            this.buildConsumer(messageConfig);
        }
        log.info("[end]初始化启动RocketMQ消费者！");
    }

    public void subscribe(MessageConfig messageConfig) {
        try {
            this.buildConsumer(messageConfig);
        } catch (Exception e) {
            log.error("消息队列[{}]订阅失败", messageConfig.getName(), e);
        }
    }

    public void unsubscribe(MessageConfig messageConfig) {
        Consumer consumer = consumerMap.get(messageConfig.getId());
        if(consumer != null) {
            String[] topicList = messageConfig.getTopic().split(",");
            for(String topic : topicList) {
                topicMessageConfigMap.remove(messageConfig.getTopic());
                consumer.unsubscribe(topic);
            }
            consumer.shutdown();
            consumerMap.remove(consumer);
        }
        log.info("取消订阅RocketMQ消费者[{}]！", messageConfig.getGroupId());
    }

    /**
     * 初始化消费者
     *
     * @return
     */
    public Consumer getConsumer(Integer id) {
        return consumerMap.get(id);
    }

    /**
     * 构建单个消费者
     */
    private void buildConsumer(MessageConfig messageConfig) {
        if(!(StringUtils.hasLength(messageConfig.getAccessKey()) && StringUtils.hasLength(messageConfig.getSecretKey())
                && StringUtils.hasLength(messageConfig.getNamesrvAddr()) && StringUtils.hasLength(messageConfig.getGroupId())
                && StringUtils.hasLength(messageConfig.getTopic()))) {
            log.error("消息队列[{}]rocketMq配置信息有误", messageConfig.getName());
            return;
        }
        Properties properties = new Properties();
        // AccessKey ID阿里云身份验证，在阿里云服务器管理控制台创建。
        properties.setProperty(PropertyKeyConst.AccessKey, messageConfig.getAccessKey());
        // AccessKey Secret阿里云身份验证，在阿里云服务器管理控制台创建。
        properties.setProperty(PropertyKeyConst.SecretKey, messageConfig.getSecretKey());
        // 设置TCP协议接入点，进入控制台的实例详情页面的TCP协议客户端接入点区域查看。
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, messageConfig.getNamesrvAddr());
        // 您在控制台创建的Group ID。
        properties.setProperty(PropertyKeyConst.GROUP_ID, messageConfig.getGroupId());
        //将消费者线程数固定为20个 20为默认值
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        // 集群订阅方式（默认）。
//         properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        // 广播订阅方式。
//         properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        // producer 实例配置初始化
        Consumer consumer = ONSFactory.createConsumer(properties);
        String[] topicList = messageConfig.getTopic().split(",");
        for(String topic : topicList) {
            messageConfig.setTopic(topic);
            topicMessageConfigMap.put(messageConfig.getTopic(), messageConfig);
            consumer.subscribe(topic, StringUtils.hasLength(messageConfig.getTag()) ? messageConfig.getTag() : "*", new RocketMQListener());
        }
        // 在发送消息前，必须调用start方法来启动consumer，只需调用一次即可，当项目关闭时，自动shutdown
        consumer.start();
        consumerMap.put(messageConfig.getId(), consumer);
        log.info("初始化RocketMQ消费者[{}]！", messageConfig.getGroupId());
    }

}
